package com.sbc.mapreduce.outputformat;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

/**
 * @description:
 * @author: songbaicheng
 * @create: 2024-07-03 15:49
 **/
public class wwwRecordWriter extends RecordWriter<Text, NullWritable> {

    private FSDataOutputStream baiduOut;
    private FSDataOutputStream otherOut;

    public wwwRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException {
        try (FileSystem fileSystem = FileSystem.get(taskAttemptContext.getConfiguration())) {
            baiduOut = fileSystem.create(new Path("/Users/songbaicheng/temp/output/www.baidu.out"));
            otherOut = fileSystem.create(new Path("/Users/songbaicheng/temp/output/www.other.out"));
        }

    }

    @Override
    public void write(Text text, NullWritable nullWritable) throws IOException, InterruptedException {

        String line = text.toString();
        if (line.contains("baidu")) {
            baiduOut.writeBytes(line + "\n");
        } else {
            otherOut.writeBytes(line + "\n");
        }
    }

    @Override
    public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {

        if (baiduOut != null) {
            baiduOut.close();
        }
        if (otherOut != null) {
            otherOut.close();
        }
    }
}
